package org.example.liuy.service.session;


import lombok.extern.slf4j.Slf4j;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicLong;

/**
 * 智能队列管理器 - 处理极端高并发场景
 */
@Slf4j
public class IntelligentQueueManager {

    // 队列配置
    private static final int HIGH_PRIORITY_QUEUE_SIZE = 1000;
    private static final int NORMAL_PRIORITY_QUEUE_SIZE = 10000;
    private static final int LOW_PRIORITY_QUEUE_SIZE = 100000;

    // 优先级队列
    private final PriorityBlockingQueue<SessionRequest> highPriorityQueue;
    private final PriorityBlockingQueue<SessionRequest> normalPriorityQueue;
    private final PriorityBlockingQueue<SessionRequest> lowPriorityQueue;

    // 处理线程池
    private final ThreadPoolExecutor processingExecutor;

    // 统计信息
    private final AtomicLong totalRequests = new AtomicLong(0);
    private final AtomicLong processedRequests = new AtomicLong(0);
    private final AtomicLong droppedRequests = new AtomicLong(0);

    // 系统负载监控
    private final AtomicLong currentLoad = new AtomicLong(0);

    public IntelligentQueueManager() {
        // 初始化优先级队列
        this.highPriorityQueue = new PriorityBlockingQueue<>(HIGH_PRIORITY_QUEUE_SIZE);
        this.normalPriorityQueue = new PriorityBlockingQueue<>(NORMAL_PRIORITY_QUEUE_SIZE);
        this.lowPriorityQueue = new PriorityBlockingQueue<>(LOW_PRIORITY_QUEUE_SIZE);

        // 初始化处理线程池
        this.processingExecutor = new ThreadPoolExecutor(
                10, // 核心线程数
                100, // 最大线程数
                60L, TimeUnit.SECONDS,
                new LinkedBlockingQueue<>(1000),
                new ThreadFactory() {
                    private final AtomicLong threadNumber = new AtomicLong(1);
                    @Override
                    public Thread newThread(Runnable r) {
                        Thread t = new Thread(r, "SessionProcessor-" + threadNumber.getAndIncrement());
                        t.setDaemon(true);
                        return t;
                    }
                }
        );

        // 启动处理任务
        startProcessing();
    }

    /**
     * 提交会话请求
     */
    public CompletableFuture<String> submitSessionRequest(Integer customerId, int priority) {
        totalRequests.incrementAndGet();

        // 检查队列容量，避免内存溢出
        if (getTotalQueueSize() > getMaxQueueSize()) {
            droppedRequests.incrementAndGet();
            log.warn("Queue overflow, dropping request for customer: {}", customerId);
            CompletableFuture<String> future = new CompletableFuture<>();
            future.complete("REJECTED:QUEUE_OVERFLOW");
            return future;
        }

        SessionRequest request = new SessionRequest(customerId, priority, System.currentTimeMillis());
        CompletableFuture<String> future = new CompletableFuture<>();
        request.setFuture(future);

        // 根据优先级放入对应队列
        switch (priority) {
            case 1: // 高优先级
                if (highPriorityQueue.size() < HIGH_PRIORITY_QUEUE_SIZE) {
                    highPriorityQueue.offer(request);
                } else {
                    droppedRequests.incrementAndGet();
                    future.complete("REJECTED:HIGH_PRIORITY_QUEUE_FULL");
                }
                break;
            case 2: // 普通优先级
                if (normalPriorityQueue.size() < NORMAL_PRIORITY_QUEUE_SIZE) {
                    normalPriorityQueue.offer(request);
                } else {
                    // 降级到低优先级队列
                    if (lowPriorityQueue.size() < LOW_PRIORITY_QUEUE_SIZE) {
                        lowPriorityQueue.offer(request.withPriority(3));
                    } else {
                        droppedRequests.incrementAndGet();
                        future.complete("REJECTED:NORMAL_PRIORITY_QUEUE_FULL");
                    }
                }
                break;
            case 3: // 低优先级
                if (lowPriorityQueue.size() < LOW_PRIORITY_QUEUE_SIZE) {
                    lowPriorityQueue.offer(request);
                } else {
                    droppedRequests.incrementAndGet();
                    future.complete("REJECTED:LOW_PRIORITY_QUEUE_FULL");
                }
                break;
        }

        return future;
    }

    /**
     * 启动处理任务
     */
    private void startProcessing() {
        // 启动多个处理线程
        for (int i = 0; i < 5; i++) {
            processingExecutor.submit(this::processRequests);
        }

        // 启动负载监控
        processingExecutor.submit(this::monitorLoad);
    }

    /**
     * 处理请求
     */
    private void processRequests() {
        while (!Thread.currentThread().isInterrupted()) {
            try {
                SessionRequest request = null;

                // 优先处理高优先级队列
                if (!highPriorityQueue.isEmpty()) {
                    request = highPriorityQueue.poll(100, TimeUnit.MILLISECONDS);
                } else if (!normalPriorityQueue.isEmpty()) {
                    request = normalPriorityQueue.poll(100, TimeUnit.MILLISECONDS);
                } else if (!lowPriorityQueue.isEmpty()) {
                    request = lowPriorityQueue.poll(1000, TimeUnit.MILLISECONDS); // 低优先级等待更长时间
                }

                if (request != null) {
                    processRequest(request);
                }

            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                break;
            } catch (Exception e) {
                log.error("Error processing session request", e);
            }
        }
    }

    /**
     * 处理单个请求
     */
    private void processRequest(SessionRequest request) {
        try {
            currentLoad.incrementAndGet();

            // 模拟会话创建过程
            String sessionKey = createSession(request.getCustomerId());

            // 完成future
            request.getFuture().complete(sessionKey);
            processedRequests.incrementAndGet();

        } catch (Exception e) {
            request.getFuture().completeExceptionally(e);
            log.error("Error processing request for customer: {}", request.getCustomerId(), e);
        } finally {
            currentLoad.decrementAndGet();
        }
    }

    /**
     * 模拟会话创建
     */
    private String createSession(Integer customerId) {
        // 这里应该调用实际的会话创建逻辑
        // 为了演示，我们简单模拟一下
        try {
            // 模拟处理时间
            Thread.sleep(10);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }

        return "session_" + customerId + "_" + System.currentTimeMillis();
    }

    /**
     * 监控系统负载
     */
    private void monitorLoad() {
        ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor();
        scheduler.scheduleAtFixedRate(() -> {
            long load = currentLoad.get();
            long total = totalRequests.get();
            long processed = processedRequests.get();
            long dropped = droppedRequests.get();

            log.info("Queue Manager Status - Load: {}, Total: {}, Processed: {}, Dropped: {}",
                    load, total, processed, dropped);

            // 动态调整线程池大小
            adjustThreadPoolSize(load);

        }, 10, 10, TimeUnit.SECONDS);
    }

    /**
     * 动态调整线程池大小
     */
    private void adjustThreadPoolSize(long currentLoad) {
        int currentCorePoolSize = processingExecutor.getCorePoolSize();

        if (currentLoad > 50 && currentCorePoolSize < 50) {
            processingExecutor.setCorePoolSize(Math.min(50, currentCorePoolSize + 5));
            log.info("Increased core pool size to: {}", processingExecutor.getCorePoolSize());
        } else if (currentLoad < 10 && currentCorePoolSize > 10) {
            processingExecutor.setCorePoolSize(Math.max(10, currentCorePoolSize - 5));
            log.info("Decreased core pool size to: {}", processingExecutor.getCorePoolSize());
        }
    }

    /**
     * 获取总队列大小
     */
    private int getTotalQueueSize() {
        return highPriorityQueue.size() + normalPriorityQueue.size() + lowPriorityQueue.size();
    }

    /**
     * 获取最大队列大小
     */
    private int getMaxQueueSize() {
        return HIGH_PRIORITY_QUEUE_SIZE + NORMAL_PRIORITY_QUEUE_SIZE + LOW_PRIORITY_QUEUE_SIZE;
    }

    /**
     * 会话请求类
     */
    private static class SessionRequest implements Comparable<SessionRequest> {
        private final Integer customerId;
        private final int priority;
        private final long timestamp;
        private CompletableFuture<String> future;

        public SessionRequest(Integer customerId, int priority, long timestamp) {
            this.customerId = customerId;
            this.priority = priority;
            this.timestamp = timestamp;
        }

        public SessionRequest withPriority(int newPriority) {
            return new SessionRequest(this.customerId, newPriority, this.timestamp);
        }

        // 按优先级排序，优先级数字越小优先级越高
        @Override
        public int compareTo(SessionRequest other) {
            int priorityCompare = Integer.compare(this.priority, other.priority);
            if (priorityCompare != 0) {
                return priorityCompare;
            }
            // 相同优先级按时间排序，先到先处理
            return Long.compare(this.timestamp, other.timestamp);
        }

        // Getters
        public Integer getCustomerId() { return customerId; }
        public int getPriority() { return priority; }
        public long getTimestamp() { return timestamp; }
        public CompletableFuture<String> getFuture() { return future; }
        public void setFuture(CompletableFuture<String> future) { this.future = future; }
    }

    /**
     * 关闭队列管理器
     */
    public void shutdown() {
        processingExecutor.shutdown();
        try {
            if (!processingExecutor.awaitTermination(60, TimeUnit.SECONDS)) {
                processingExecutor.shutdownNow();
            }
        } catch (InterruptedException e) {
            processingExecutor.shutdownNow();
            Thread.currentThread().interrupt();
        }
    }
}
